KAFKA-14365: Extract common logic from Fetcher#13425
Conversation
|
|
Extract logic from Fetcher into AbstractFetcher. Also introduce FetchConfig as a more concise way to delineate state from incoming configuration. Formalized the defaults in CommonClientConfigs and ConsumerConfig to be accessible elsewhere.
|
@guozhangwang @hachikuji @rajinisivaram @philipnee This is ready for a review from whomever has the time. |
guozhangwang
left a comment
guozhangwang
left a comment
There was a problem hiding this comment.
Overall looks good to me. Except the question for synchronizing the super's functions I think we can move forward to merge.
| config.getBoolean(ConsumerConfig.THROW_ON_FETCH_STABLE_OFFSET_UNSUPPORTED), | ||
| config.getString(ConsumerConfig.CLIENT_RACK_CONFIG)); | ||
| } | ||
| FetchConfig<K, V> fetchConfig = new FetchConfig<>(config, keyDeserializer, valueDeserializer, isolationLevel); |
There was a problem hiding this comment.
This is a nice improvement, +1!
There was a problem hiding this comment.
| FetchConfig<K, V> fetchConfig = new FetchConfig<>(config, keyDeserializer, valueDeserializer, isolationLevel); | |
| FetchConfig<K, V> fetchConfig = new FetchConfig<>(config, this.keyDeserializer, this.valueDeserializer, isolationLevel); |
There was a problem hiding this comment.
@kirktrue , these should be this.keyDeserializer and this.valueDeserializer. Otherwise the original values may be null and will cause an NPE on line 300 of CompletedFetch.java
| * | ||
| * @return true if there are completed fetches, false otherwise | ||
| */ | ||
| boolean hasCompletedFetches() { |
There was a problem hiding this comment.
I made a quick browse on this class assuming that all the logic are not changed, i.e. mostly copy-paste here except the FetchConfig wrapping. Please LMK otherwise and I will give it a deeper look.
There was a problem hiding this comment.
I intentionally did my best to keep the code I moved from Fetcher into AbstractFetch identical. There was some minor refactoring which could technically introduce bugs.
|
I've checked the failed tests, which are irrelevant. |
|
LGTM, merged to trunk. |
|
@guozhangwang , @kirktrue , this PR is causing NPEs downstream. I've added a comment above that indicates the problematic lines. Thanks. |
philipnee
added
KIP-848
5 participants
The
Fetcherclass is used internally by theKafkaConsumerto fetch records from the brokers. There is ongoing work to create a new consumer implementation with a significantly refactored threading model. The threading refactor work requires a similarly refactoredFetcher.This task includes refactoring
Fetcherby extracting out some common logic to allow forthcoming implementations to leverage it.FetcherintoAbstractFetchFetchConfigas a concise way to delineate incoming configuration from stateCommonClientConfigsandConsumerConfigto be accessible elsewhereCommitter Checklist (excluded from commit message)